/**
 * @file StreamAnalyser.c
 *
 * This file implements the StreamAnalyser interface.
 *
 * @author Luc Fourestier
 * @date 2009
 *
 * @page SA The StreamAnalyzer
 *
 * The Stream Analyzer receives a stream, buffer it into a circular buffer then
 * parse it and treat it according to the options (Send it back on UDP, dump it to a file...)
 *
 * The StreamAnalyzer watches the chunks out of the circular buffer and tags them with proper timestamps
 * when to send it via the StreamPlayer.
 *
 * @section SA_Analysis The analysis/tagging algorithm
 * The stream analysis has two indexes on the circular buffer: The analysis index which keeps track where
 * the analysis is, and the tagging index which keeps tracks where the tagging is.
 * Indeed, in order to calculate the proper timestamp when to send the chunks, wee need
 * to know always two PCRs in advance compared to the player. Thus, the analysis index
 * advances until it finds a PCR, it then triggers the tagging algo (which was waiting),
 * The tagging algo can then calculate the timestamp and will tag all packets until it
 * gets back at the same index as the analysis.
 *
 * @subsection SA_Tag The tag timestamp calculation
 * Till we do not have a PCR, no tags are done.
 * At the first PCR, as we still don't know the bitrate (two PCRs are required for that!), we stamps
 * the previous chunks with 0, which will tells to the player to send them immediately.
 * At the second PCRs and the other next, we know the bitrate now, we can then calculate the proper stamps
 * for the chunks which are still waiting for tags. We tag them all at once then. And go back to the
 * analysis for the next chunks to come.
 * The calculation is on a per PCR period basis. We do calculate the bitrate on a PCR period (in between
 * two PCRs), then all the chunks in between this two PCRS are then regularly stamped (spread all over the
 * period).
 *
 * TO BE COMPLETED
 *
 */

#include "StreamAnalyser.h"

#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <math.h>

#include "TS.h"
#include "PCR.h"

// Defines
#define MAGIC 0x55
#define GUARD_TIME_US 50 //!< Used in the jitter algorithm to save CPU
#define SOCKET_SNDBUFF_SIZE 150000

// Statics
static void * AnalyzerThread(void * Arg);
static int AnalyzeStream(SA_StreamAnalyzer_t * SA, unsigned char *Buffer, unsigned int *Size);

/**
 * Create a stream analyzer
 *
 * @param SA (out): The created stream analyzer
 * @param Options (in): Options to follow while analyzing
 * @return Errors
 */
int SA_Create(SA_StreamAnalyzer_t ** SA_p, SA_Option_t Options)
{
	int RetVal = OK;
	SA_StreamAnalyzer_t * SA;

	if (SA_p != NULL)
	{
		// Create the SA
		SA = (SA_StreamAnalyzer_t*) malloc(sizeof(SA_StreamAnalyzer_t));
		if (SA == NULL)
		{
			return(SA_ERROR_CANNOT_ALLOCATE);
		}
		bzero((void*)SA, sizeof(SA_StreamAnalyzer_t));

		// remember options
		SA->Options = Options;

		// Init jitter record
#ifdef DEBUG_JITTER_FILE
		SA->JitterFile = fopen(SA->Options.JitterFileName, "w+");
		if (SA->JitterFile != NULL)
		{
			fprintf(SA->JitterFile, "Packet;Timestamp;PCR\n");
		}
		else
		{
			return(SA_ERROR_CANNOT_OPEN_FILE);
		}
#endif

		// Allocate circular buffers
		RetVal = CircB_CreateCircBuff(&SA->CircBuffer, SA->Options.MaxBufferSize+sizeof(Tag_t), SA->Options.MaxBufferNumber);
		if (RetVal != OK)
		{
			return(SA_ERROR_CANNOT_ALLOCATE);
		}

		SA->Magic = MAGIC;

		// Init the analyzer thread
		SA->ThreadLoop = 1;
		if (pthread_create(&SA->AnalyzerThreadID, NULL, AnalyzerThread, (void*) SA) != 0)
		{
			return(SA_ERROR_CANNOT_ALLOCATE);
		}

		// Return the structure
		*SA_p = SA;
	}
	else
	{
		RetVal = SA_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * Delete the stream analyzer
 *
 * @param SA (in): The stream analyzer
 * @return Errors
 */
int SA_Delete(SA_StreamAnalyzer_t * SA)
{
	int RetVal = OK;

	if (SA != NULL)
	{
		// Kill threads
		SA->ThreadLoop = 0;
		CircB_KillWrite(SA->CircBuffer);
		CircB_KillAnalysis(SA->CircBuffer);
		CircB_KillTag(SA->CircBuffer);
		CircB_KillRead(SA->CircBuffer);
		pthread_join(SA->AnalyzerThreadID, NULL);

		// Clear the magic number (SA cannot be miss-used anymore then)
		SA->Magic = 0;

		// Delete circular buffers
		RetVal = CircB_DeleteCircBuff(SA->CircBuffer);

		// Close jitter file
#ifdef DEBUG_JITTER_FILE
		if (SA->JitterFile != NULL)
		{
			fclose(SA->JitterFile);
			SA->JitterFile = NULL;
		}
#endif

		free(SA);
	}
	else
	{
		RetVal = SA_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * Print the StreamAnalyzer info structure
 *
 * @param SA (in): The SA to print
 * @param Name (in): The name of the SA (any nickname you wish!).
 * @param level (in): Level of details (Lowest = 0).
 * @param Output (in): Where to print (stdout or FILE).
 * @return Errors
 */
int SA_Print(SA_StreamAnalyzer_t * SA, char * Name, int Level, FILE * Output)
{
	int RetVal = OK;

	if ((SA == NULL) || (SA->Magic != MAGIC) || (Output == NULL))
	{
		return(SA_ERROR_BAD_PARAM);
	}

	if (Name == NULL)
	{
		Name = "StreamAnalyzer";
	}

	fprintf(Output, "%s duration: %.6f\n", Name, SA->Duration);
	fprintf(Output, "%s average bitrate: %i\n", Name, SA->AvgBitrate);
	fprintf(Output, "%s total packet number: %lli\n", Name, SA->TotalPacketNumber);
	fprintf(Output, "%s total bytes analyzed: %lli\n", Name, SA->TotalPacketNumber*TS_SIZE);

	// Intermediate status
	if (Level >= PRINT_LEVEL_INTERMEDIATE)
	{
		fprintf(Output, "%s instant bitrate: %i\n", Name, SA->InstantBitrate);
		fprintf(Output, "%s total packet error: %lli\n", Name, SA->TotalPacketError);
		fprintf(Output, "%s total PAT number: %lli\n", Name, SA->TotalPATNumber);
		fprintf(Output, "%s total PMT number: %lli\n", Name, SA->TotalPMTNumber[0]);
		fprintf(Output, "%s total PCR number: %lli\n", Name, SA->TotalPCRNumber);
	}

	// Internals status
	if (Level >= PRINT_LEVEL_INTERNALS)
	{
		//	fprintf(Output, "%s bitrate deviation: %i\n", Name, SA->BRDeviation);
		fprintf(Output, "%s buffer length: %i\n", Name, SA->CircBuffer->MaxElementNumber);
		fprintf(Output, "%s buffer usage: %i\n", Name, SA->CircBuffer->ElementCount);
		fprintf(Output, "%s buffer element size: %i\n", Name, SA->Options.MaxBufferSize);
		fprintf(Output, "%s threads state: %i\n", Name, (int)SA->ThreadLoop);
		fprintf(Output, "%s total circular Buffer error: %lli\n", Name, SA->TotalCircError);
		fprintf(Output, "%s last PCR: %lli\n", Name, SA->LastPCR);
		fprintf(Output, "%s new PCR: %lli\n", Name, SA->NewPCR);

#ifdef DEBUG_SA_PROFILING
		if (SA->passCount > 1)
		{
			fprintf(Output, "%s total pass duration: %is %iµs\n", Name, (int)SA->TotalPassDuration.tv_sec, (int)SA->TotalPassDuration.tv_usec);
			fprintf(Output, "%s pass count: %lli\n", Name, SA->passCount-1);
			fprintf(Output, "%s Average duration per pass: %lli µs\n", Name, TIMEVAL_TO_USECONDS(&SA->TotalPassDuration)/(SA->passCount-1));
			fprintf(Output, "%s pass errors: %lli\n", Name, SA->passError);
		}
#endif
	}

	// Full status
	if (Level >= PRINT_LEVEL_FULL)
	{
		unsigned int i = 0;
		char PMTName[SA_MAX_NAME_LENGTH];

		fprintf(Output, "\n");
		PAT_Print(&SA->PAT, NULL, Level, Output);
		for (i=0; (i<SA->PAT.ProgramDescNumber) && (i<PAT_MAX_PROGRAM_DESC); i++)
		{
			// Find the first program
			if (SA->PAT.program_descriptors[i].program_number != 0)
			{
				break;
			}
		}
		snprintf(PMTName,SA_MAX_NAME_LENGTH, "PMT[%i]", i);
		PMT_Print(&SA->PMT[i], PMTName, Level, Output);
	}


	fprintf(Output, "\n");

	return(RetVal);
}

/**
 * Analyze the stream, chunk after chunk. Keep track of the position in the stream.
 * Do TS packet, PAT, PMT, and time analysis.
 *
 * @param SA (in): The stream analyzer
 * @param Buffer (in): The new chunk to treat
 * @param Size (in/out): The size of the chunk buffer, then the actual analyzed size
 * @param Tag (out): The tagging for this chunk
 * @return Errors
 */
static int AnalyzeStream(SA_StreamAnalyzer_t * SA, unsigned char *Buffer, unsigned int *Size)
{
	int RetVal = OK;
	unsigned int Index = 0;
	unsigned int TempSize = *Size;
	TS_t TSPacket;
	unsigned int i = 0;

	if ((SA != NULL) && (SA->Magic == MAGIC) && (Buffer != NULL) && (Size != NULL) && (*Size >= TS_SIZE))
	{
		// Analyze buffer
		while (Index < *Size)
		{
			// Get TS packet
			TempSize = *Size - Index;
			RetVal = TS_Create(Buffer+Index, &TempSize, &TSPacket);
			if (RetVal == OK)
			{
				Index += TempSize;
				SA->TotalPacketNumber++;
			}
			else if (RetVal == TS_ERROR_NOT_A_TS)
			{
				SA->TotalPacketError++;
				SA->TotalPacketNumber++;
				Index += TS_SIZE; // try to continue anyway
				RetVal = OK;
				continue;
			}
			else
			{
				RetVal = SA_ERROR_CANNOT_ANALYZE;
				break;
			}

			// Find PAT
			if (TSPacket.PID == PAT_PID)
			{
				int Ret = 0;
				unsigned int DataSize = TSPacket.DataByteLength;
				if (SA->TotalPATNumber == 0) // No PAT already found, create one!
				{
					Ret = PAT_Create(TSPacket.data_byte+1, &DataSize, &SA->PAT);
					if (Ret == OK)
					{
						SA->TotalPATNumber++;
					}
				}
				else // PAT already found, just verify that this one is the same
				{
					unsigned short TSID = 0xffff;
					Ret = PAT_GetTSID(TSPacket.data_byte+1, DataSize, &TSID);
					if (Ret == OK)
					{
						if (TSID == SA->PAT.transport_stream_id)
						{
							SA->TotalPATNumber++;
						}
					}
				}
			}

			// Find PMT
			if (SA->TotalPATNumber > 0)
			{
				// Check if PID is a PMT
				for (i=0; (i<SA->PAT.ProgramDescNumber) && (i<PAT_MAX_PROGRAM_DESC); i++)
				{
					if ((SA->PAT.program_descriptors[i].program_number != 0) &&
						(TSPacket.PID == SA->PAT.program_descriptors[i].PID))
					{
						break;
					}
				}

				if ((i < SA->PAT.ProgramDescNumber) && (i < PAT_MAX_PROGRAM_DESC)) // PID is a PMT
				{
					int Ret = 0;
					unsigned int DataSize = TSPacket.DataByteLength;
					if (SA->TotalPMTNumber[i] == 0) // No PMT found yet
					{
						PMT_t PMT;
						Ret = PMT_Create(TSPacket.data_byte+1, &DataSize, &PMT);
						if (Ret == OK)
						{
							SA->PMT[i] = PMT;
							SA->TotalPMTNumber[i]++;
						}
					}
					else
					{
						Ret = PMT_CheckPMT(TSPacket.data_byte+1, DataSize);
						if (Ret == OK)
						{
							SA->TotalPMTNumber[i]++;
						}
					}
				}
			}

			// Filter out PCRs
			for (i=0; (i<SA->PAT.ProgramDescNumber) && (i<PAT_MAX_PROGRAM_DESC); i++)
			{
				// Find the first program
				// We will rely on the first program PCR only
				if (SA->PAT.program_descriptors[i].program_number != 0)
				{
					break;
				}
			}

			if (SA->TotalPMTNumber[i] > 0)
			{
				if ((TSPacket.PID == SA->PMT[i].PCR_PID) &&
					(TSPacket.adaptation_field_control & 0x2) &&
					(TSPacket.adaptation_field.adaptation_field_length > 0) &&
					(TSPacket.adaptation_field.PCR_flag == 1))
				{
					unsigned long long PCR = PCR_TO_PCR64(TSPacket.adaptation_field.program_clock_reference_base, TSPacket.adaptation_field.program_clock_reference_extension);

					SA->NewPCRFound = 1;

					// Remember new and last PCRs
					SA->LastPCR = SA->NewPCR;
					SA->LastPCRPckNumber = SA->NewPCRPckNumber;
					SA->LastPCRTime = SA->NewPCRTime;
					SA->LastPCROffset = SA->NewPCROffset;

					SA->NewPCR = PCR;
					SA->NewPCRPckNumber = SA->TotalPacketNumber; // Can never be zero (because wee need at least a PMT packet)
					SA->NewPCROffset = Index/TS_SIZE -1;

					// update PCR difference
					if (SA->NewPCR >= SA->LastPCR) // No wrap
					{
						SA->PCRDelta = SA->NewPCR - SA->LastPCR;
					}
					else // PCR has wrapped
					{
						SA->PCRDelta = PCR64_WRAP + SA->NewPCR - SA->LastPCR;
					}

					SA->TotalPCRNumber++;

					// Calculate duration and bitrate
					if (SA->LastPCRPckNumber != 0) // Only if two PCRs have been found (therefore if last PCR packet number is not null)
					{
						SA->InstantBitrate = (unsigned int)(((double)(SA->NewPCRPckNumber-SA->LastPCRPckNumber) * TS_SIZE * 8) / (PCR64_TO_S((double)SA->PCRDelta)));

						// If B is the new instant bitrate, M the mean value then Q the deviation is: Qn = Qn-1 + (n - 1) / n * sqr(Bn - Mn-1)
//						SA->SqrBRDevSum +=  (int)(SA->TotalPCRNumber-2)*((int)SA->InstantBitrate-(int)SA->AvgBitrate)*((int)SA->InstantBitrate-(int)SA->AvgBitrate) / (SA->TotalPCRNumber-1);
//						SA->BRDeviation = (int)sqrt(SA->SqrBRDevSum/(SA->TotalPCRNumber-1));

						SA->Duration += PCR64_TO_S((double)SA->PCRDelta);
						SA->AvgBitrate = (unsigned int)(((double)SA->TotalPacketNumber * TS_SIZE * 8) / SA->Duration);
					}

				}
			}
		}

		// update Size
		*Size = Index;
	}
	else
	{
		RetVal = SA_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * The analyzer thread.
 * The Analysis which advance into the stream to tag the incoming chunks, using an extra circular buffer.
 *
 * @param Arg: the SA_StreamAnalyzer_t structure
 */
static void * AnalyzerThread(void * Arg)
{
	void * Buff_p = NULL;
	SA_StreamAnalyzer_t * SA = (SA_StreamAnalyzer_t *) Arg;

	int Error = OK;

	if (SA == NULL)
	{
		pthread_exit(NULL);
		return(NULL);
	}

#ifdef DEBUG_SA_PROFILING
		struct timeval Start;
		Start.tv_sec = 0;
		Start.tv_usec = 0;
#endif

	while (SA->ThreadLoop)
	{
		unsigned int Size = 0;

#ifdef DEBUG_SA_PROFILING
			struct timeval New;
			gettimeofday(&New, NULL);
			if (Start.tv_sec == 0)
			{
				Start = New;
			}
			timersub(&New, &Start, &SA->TotalPassDuration);
			SA->passCount++;
#endif

		do
		{
			// *** Analyze the stream ***

			// Get chunk
			Error = CircB_GetAnalysisElt(SA->CircBuffer, &Buff_p, &Size, CIRCB_WAIT_FOREVER);
			if (Error != OK)
			{
				break;
			}

			// Analyze it
			Size -= sizeof(Tag_t);
			Error = AnalyzeStream(SA, ((unsigned char*)Buff_p)+sizeof(Tag_t), &Size);
			if (Error != OK)
			{
				Error = CircB_ReleaseAnalysisElt(SA->CircBuffer);
				break;
			}

			// Validate the chunk
			Error = CircB_ValidateAnalysisElt(SA->CircBuffer);


			// *** Tag the stream ***

//! @todo Should the SA know about the output type?
			if (SA->Options.OutputType == File)
			{
				// Release the chunk for the player
				((Tag_t*)Buff_p)->ReadyToPlay = 1;

				Error = CircB_ValidateTagElt(SA->CircBuffer);
			}
			else if (SA->Options.OutputType == Udp)
			{
				if ((SA->NewPCRFound) || (((Tag_t*)Buff_p)->ChunkIsLast))
				{
					SA->NewPCRFound = 0;

					// Tag all the previous chunks
					unsigned int n = 1;
					unsigned int Temp = SA->CircBuffer->AnalysisIndex == 0 ?  SA->CircBuffer->MaxElementNumber-1 : SA->CircBuffer->AnalysisIndex-1;
					while (SA->CircBuffer->TagIndex != Temp)
					{
						void * TagBuff_p = NULL;
						unsigned int TagSize = 0;

						// Get the chunk
						Error = CircB_GetTagElt(SA->CircBuffer, &TagBuff_p, &TagSize, CIRCB_WAIT_FOREVER);
						if (Error != OK)
						{
							break;
						}

						// Calculate TimeToSend
						if (SA->LastPCRPckNumber == 0) // Only One PCR has been found so far
						{
							// Send immediately all previous chunks
							((Tag_t*)TagBuff_p)->TimeToSend.tv_sec = 0;
							((Tag_t*)TagBuff_p)->TimeToSend.tv_usec = 0;
						}
						else // We have two PCRs, we can calculate!
						{
							struct timeval TempTime;
							unsigned long long Temp = n * MAX_PACKET_PER_MTU * SA->PCRDelta / (SA->NewPCRPckNumber - SA->LastPCRPckNumber);
							TempTime.tv_sec = PCR64_TO_US(Temp) / 1000000;
							TempTime.tv_usec = PCR64_TO_US(Temp) % 1000000;

							timeradd(&SA->LastPCRTime, &TempTime, &((Tag_t*)TagBuff_p)->TimeToSend);
						}

						n++;

						// Release the chunk for the player
						((Tag_t*)TagBuff_p)->ReadyToPlay = 1;

						Error = CircB_ValidateTagElt(SA->CircBuffer);
					}

					// Now tag this chunk (it is mandatory a PCR!)
					if (SA->CircBuffer->TagIndex  == Temp)
					{
						// Calculate TimeToSend
						if (SA->LastPCRPckNumber == 0) // Only One PCR has been found so far
						{
							// Try to evaluate when to treat it
							struct timeval Now;
							unsigned long long Temp = SA->TotalPacketNumber / MAX_PACKET_PER_MTU * 500; // give 500 µs per previous chunk to be treated

							gettimeofday(&Now, NULL);
							timersub(&Now, &SA->Options.ReferenceTime, &Now); // Relative time
							SA->NewPCRTime.tv_sec = Temp / 1000000;
							SA->NewPCRTime.tv_usec = Temp % 1000000;
							timeradd(&SA->NewPCRTime, &Now, &SA->NewPCRTime);
						}
						else // We have two PCRs, we can calculate!
						{
							struct timeval TempTime;
							TempTime.tv_sec = PCR64_TO_US(SA->PCRDelta) / 1000000;
							TempTime.tv_usec = PCR64_TO_US(SA->PCRDelta) % 1000000;

							timeradd(&SA->LastPCRTime, &TempTime, &SA->NewPCRTime);
						}

						((Tag_t*)Buff_p)->TimeToSend.tv_sec = SA->NewPCRTime.tv_sec;
						((Tag_t*)Buff_p)->TimeToSend.tv_usec = SA->NewPCRTime.tv_usec;

#ifdef DEBUG_JITTER_FILE
						// Dump the time stamps for jitter calculation into a file
						if (SA->JitterFile != NULL)
						{
							fprintf(SA->JitterFile, "%lld;%.6f;%.6f\n", SA->TotalPacketNumber, TIMEVAL_TO_SECONDS(&((Tag_t*)Buff_p)->TimeToSend), PCR64_TO_S((double)SA->NewPCR));
						}
#endif

						// Release the chunk for the player
						((Tag_t*)Buff_p)->ReadyToPlay = 1;

						Error = CircB_ValidateTagElt(SA->CircBuffer);

						// Reset the PCRs and timestamp if chunk is last
						if (((Tag_t*)Buff_p)->ChunkIsLast)
						{
							unsigned int i = 0;

							SA->AvgBitrate = 0;
							SA->Duration = 0.0;
							SA->InstantBitrate = 0;
							SA->LastPCR = 0;
							SA->LastPCROffset = 0;
							SA->LastPCRPckNumber = 0;
							SA->LastPCRTime.tv_sec = 0;
							SA->LastPCRTime.tv_usec = 0;
							SA->NewPCR = 0;
							SA->NewPCROffset = 0;
							SA->NewPCRPckNumber = 0;
							SA->NewPCRTime.tv_sec = 0;
							SA->NewPCRTime.tv_usec = 0;
							SA->PCRDelta = 0;
							SA->TotalPCRNumber = 0;
							SA->TotalPATNumber = 0;
							for (i=0; i<SA_MAX_PROGRAM; i++)
							{
								SA->TotalPMTNumber[i] = 0;
							}
							SA->TotalPacketNumber = 0;
						}
					}
				}
			}
		} while(0); // just to be able to break in the middle (more readable)

#ifdef DEBUG_SA_PROFILING
		if (Error != OK)
		{
			SA->passError++;
		}
		Error = OK;
#endif
	} // while(ThreadLoop)

	pthread_exit(NULL);
}


